Lambda でリソースベースポリシーのプリンシパルを監視してみた

import boto3
import json
import os
from botocore.exceptions import ClientError

# AWSクライアントの初期化
s3 = boto3.client('s3')
sns = boto3.client('sns')
sqs = boto3.client('sqs')
lambda_client = boto3.client('lambda')

SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']
BUCKETS_TO_MONITOR = os.environ.get('BUCKETS_TO_MONITOR', '').split(',')

# オプション: リージョンの指定が必要な場合
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')  # 例: 'ap-northeast-1'

def get_s3_buckets():
    try:
        response = s3.list_buckets()
        return [bucket['Name'] for bucket in response['Buckets']]
    except ClientError as e:
        print(f"Error listing S3 buckets: {e}")
        return []

def get_s3_bucket_policies(buckets):
    policies = {}
    for bucket in buckets:
        try:
            policy = s3.get_bucket_policy(Bucket=bucket)
            policies[bucket] = json.loads(policy['Policy'])
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'NoSuchBucketPolicy':
                policies[bucket] = {}
            elif error_code == 'NoSuchBucket':
                print(f"バケット '{bucket}' が存在しません。")
                policies[bucket] = {}
            else:
                print(f"バケット '{bucket}' のポリシー取得中にエラーが発生しました: {e}")
                policies[bucket] = {}
        except Exception as e:
            print(f"バケット '{bucket}' のポリシー取得中に予期しないエラーが発生しました: {e}")
            policies[bucket] = {}
    return policies

def get_sns_policies():
    policies = {}
    try:
        response = sns.list_topics()
        topics = response.get('Topics', [])
        for topic in topics:
            topic_arn = topic['TopicArn']
            try:
                policy = sns.get_topic_attributes(TopicArn=topic_arn)['Attributes'].get('Policy', '{}')
                policies[topic_arn] = json.loads(policy)
            except ClientError as e:
                error_code = e.response['Error']['Code']
                if error_code == 'NotFound':
                    policies[topic_arn] = {}
                else:
                    print(f"SNSトピック '{topic_arn}' のポリシー取得中にエラーが発生しました: {e}")
                    policies[topic_arn] = {}
            except Exception as e:
                print(f"SNSトピック '{topic_arn}' のポリシー取得中に予期しないエラーが発生しました: {e}")
                policies[topic_arn] = {}
    except ClientError as e:
        print(f"Error listing SNS topics: {e}")
    return policies

def get_sqs_policies():
    policies = {}
    try:
        response = sqs.list_queues()
        queues = response.get('QueueUrls', [])
        for queue_url in queues:
            try:
                attributes = sqs.get_queue_attributes(QueueUrl=queue_url, AttributeNames=['Policy'])['Attributes']
                policy = attributes.get('Policy', '{}')
                policies[queue_url] = json.loads(policy)
            except ClientError as e:
                error_code = e.response['Error']['Code']
                if error_code == 'AWS.SimpleQueueService.NonExistentQueue':
                    print(f"キュー '{queue_url}' が存在しません。")
                    policies[queue_url] = {}
                else:
                    print(f"SQSキュー '{queue_url}' のポリシー取得中にエラーが発生しました: {e}")
                    policies[queue_url] = {}
            except Exception as e:
                print(f"SQSキュー '{queue_url}' のポリシー取得中に予期しないエラーが発生しました: {e}")
                policies[queue_url] = {}
    except ClientError as e:
        print(f"Error listing SQS queues: {e}")
    return policies

def get_lambda_policies():
    policies = {}
    try:
        response = lambda_client.list_functions()
        functions = response.get('Functions', [])
        for function in functions:
            function_name = function['FunctionName']
            try:
                policy_response = lambda_client.get_policy(FunctionName=function_name)
                policy = json.loads(policy_response['Policy'])
                policies[function_name] = policy
            except ClientError as e:
                error_code = e.response['Error']['Code']
                if error_code == 'ResourceNotFoundException':
                    policies[function_name] = {}
                else:
                    print(f"Lambda関数 '{function_name}' のポリシー取得中にエラーが発生しました: {e}")
                    policies[function_name] = {}
            except Exception as e:
                print(f"Lambda関数 '{function_name}' のポリシー取得中に予期しないエラーが発生しました: {e}")
                policies[function_name] = {}
    except ClientError as e:
        print(f"Error listing Lambda functions: {e}")
    return policies

def check_principal_arns(policy_dict, resource_type):
    problematic_resources = []
    for resource, policy in policy_dict.items():
        if 'Statement' in policy:
            statements = policy['Statement']
            if not isinstance(statements, list):
                statements = [statements]
            for stmt in statements:
                principals = []
                if 'Principal' in stmt:
                    principal = stmt['Principal']
                    if isinstance(principal, dict):
                        for key, value in principal.items():
                            if isinstance(value, str):
                                principals.append(value)
                            elif isinstance(value, list):
                                principals.extend(value)
                    elif isinstance(principal, str):
                        principals.append(principal)
                for principal_arn in principals:
                    if isinstance(principal_arn, str) and (principal_arn.startswith('AROA') or principal_arn.startswith('AIDA')):
                        # 管理コンソールURLの生成
                        if resource_type == 's3':
                            url = f"https://s3.console.aws.amazon.com/s3/buckets/{resource}/permissions?tab=permissions"
                        elif resource_type == 'sns':
                            url = f"https://console.aws.amazon.com/sns/v3/home?region={AWS_REGION}#/topics/{resource}/permissions"
                        elif resource_type == 'sqs':
                            url = f"https://console.aws.amazon.com/sqs/v2/home?region={AWS_REGION}#/queue/{resource}/permissions"
                        elif resource_type == 'lambda':
                            url = f"https://console.aws.amazon.com/lambda/home?region={AWS_REGION}#/functions/{resource}/permissions"
                        else:
                            url = ""
                        problematic_resources.append({
                            'ResourceType': resource_type.upper(),
                            'Resource': resource,
                            'PrincipalARN': principal_arn,
                            'Policy': policy,
                            'URL': url
                        })
    return problematic_resources

def lambda_handler(event, context):
    all_problems = []

    # 監視対象のバケットを取得
    if BUCKETS_TO_MONITOR and BUCKETS_TO_MONITOR != ['']:
        s3_buckets = BUCKETS_TO_MONITOR
    else:
        s3_buckets = get_s3_buckets()

    # S3ポリシー取得とチェック
    s3_policies = get_s3_bucket_policies(s3_buckets)
    s3_problems = check_principal_arns(s3_policies, 's3')
    all_problems.extend(s3_problems)

    # SNSポリシー取得とチェック
    sns_policies = get_sns_policies()
    sns_problems = check_principal_arns(sns_policies, 'sns')
    all_problems.extend(sns_problems)

    # SQSポリシー取得とチェック
    sqs_policies = get_sqs_policies()
    sqs_problems = check_principal_arns(sqs_policies, 'sqs')
    all_problems.extend(sqs_problems)

    # Lambdaポリシー取得とチェック
    lambda_policies = get_lambda_policies()
    lambda_problems = check_principal_arns(lambda_policies, 'lambda')
    all_problems.extend(lambda_problems)

    if all_problems:
        message = "以下のリソースでプリンシパルに'AROA'または'AIDA'で始まるARNが検出されました。\n\n"
        for problem in all_problems:
            message += (
                f"リソースタイプ: {problem['ResourceType']}\n"
                f"リソース名: {problem['Resource']}\n"
                f"プリンシパルARN: {problem['PrincipalARN']}\n"
                f"ポリシー: {json.dumps(problem['Policy'], indent=2)}\n"
                f"管理コンソールURL: {problem['URL']}\n\n"
            )

        # SNSで通知
        try:
            sns.publish(
                TopicArn=SNS_TOPIC_ARN,
                Subject='リソースベースポリシーのプリンシパルにAROA/AIDAが検出されました',
                Message=message
            )
            print(f"{len(all_problems)} 件の問題を検出し、通知を送信しました。")
        except ClientError as e:
            print(f"SNSへの通知送信中にエラーが発生しました: {e}")

        return {
            'statusCode': 200,
            'body': json.dumps(f"{len(all_problems)} 件の問題を検出し、通知を送信しました。")
        }
    else:
        print("問題は検出されませんでした。")
        return {
            'statusCode': 200,
            'body': json.dumps('問題は検出されませんでした。')
        }